#include "ThreadPool.h"
#include "Bussiness.h"

void cleanup(void* arg) {
    ThreadPool* pool = (ThreadPool*)arg;
    pthread_rwlock_unlock(&pool->rwlock);
}

void* event_loop(void* arg) {
    ThreadPool* pool = (ThreadPool*) arg;
    pthread_t tid = pthread_self();
    // 线程清理函数
    pthread_cleanup_push(cleanup, pool);
    for(;;) {
        int connfd = blockq_pop(pool->task_queue); // 阻塞点

        pthread_rwlock_rdlock(&pool->rwlock);
        if (pool->exit_flag) {
            printf("%lu exits\n", tid);
            pthread_exit(0);
        }
        pthread_rwlock_unlock(&pool->rwlock);

        printf("%lu: got job\n", tid);
        // 子进程处理业务逻辑
        send_file(connfd);
        printf("%lu: job done\n", tid);
    }          
    pthread_cleanup_pop(0);
}

ThreadPool* create_thread_pool(int n)
{
    ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool));
    pool->threads = (pthread_t*) calloc(n, sizeof(pthread_t));
    pool->num_threads = n;
    pool->task_queue = blockq_create();
    pool->exit_flag = 0;
    pthread_rwlock_init(&pool->rwlock, NULL);

    // 创建线程
    for(int i = 0; i < n; i++) {
        pthread_create(&pool->threads[i], NULL, event_loop, pool);
        printf("INFO: create a thread %lu\n", pool->threads[i]);
    }
    return pool;
}

void destroy_thread_pool(ThreadPool* pool) 
{
    free(pool->threads);
    blockq_destroy(pool->task_queue);
    pthread_rwlock_destroy(&pool->rwlock);
    free(pool);
}

